<?php
/**
 * Created by PhpStorm.
 * User: BlueIdeaLirui
 * Date: 2019/11/11
 * Time: 10:13
 */

namespace business\company\services\queues;

use business\base\caches\DataCache;
use business\company\repositories\FhSupplierMulitDetailRepository;
use business\company\repositories\ProductRepository;
use business\company\repositories\SelectionSupplierProjectRepository;
use business\company\repositories\SelectionSupplierRepository;
use business\company\repositories\SelectionSupplierRevenueRepository;
use business\company\repositories\SupplierBoutiqueRepository;
use business\company\repositories\SupplierBrandRepository;
use business\company\repositories\SupplierContactRepository;
use business\company\repositories\SupplierProjectRepository;
use business\company\repositories\SupplierCategoryRepository;
use business\company\repositories\SupplierRecruitRepository;
use business\company\repositories\SupplierRepository;
use business\user\services\UserService;
use common\queue\QueueConstants;
use common\utils\DateTimeUtils;

/**
 * 平台供应商队列生产者服务
 * Class SupplierProducerService
 * @package business\company\services\queues
 */
class SupplierProducerService extends BaseService
{
    /**
     * 增量方式更新时: 记录最近一次供应商索引查询的时间的key
     */
    const LATEST_SUPPLIER_QUERY_TIME_KEY = 'Last-Company-Supplier-Query-Time';

    /**
     * 指定模块
     */
    const MODULE = 'supplier';

    /**
     * 单一更新供应商索引
     * @param $supplierId
     * @return mixed|void
     */
    public function single($supplierId)
    {
        if (!empty($supplierId)) {
            /** @var iQueue $queue */
            $queue = \yii::$app->iQueue->createClient(
                QueueConstants::COMPANY_SUPPLIER_INDEX_QUEUE
            );

            $values = implode(self::DELIMITER, [$supplierId]);
            $data  = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));
            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            $dataCache = new DataCache();

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
                && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) {
                // echo '平台供应商-已插入更新队列: ' . PHP_EOL;
            });
        }
    }

    /**
     * 全量更新精选供应商索引
     * @return mixed|void
     */
    private function fullSupplierBoutique($queue, DataCache $dataCache)
    {
        $supplierBoutiqueRepository = new SupplierBoutiqueRepository();

        $totalCount = $supplierBoutiqueRepository->getSupplierCountByDate();
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";

        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierBoutiqueRepository->getAllSupplierId($lastSupplierId,
                self::MAX_COUNT_PER_QUERY);

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values   = implode(self::DELIMITER, $changedSupplierIds);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
                && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $i) {
                echo '平台精选供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 全量更新供应商索引
     * @return mixed|void
     */
    public function full()
    {
        /** @var iQueue $queue */
        $queue = \yii::$app->iQueue->createClient(
            QueueConstants::COMPANY_SUPPLIER_INDEX_QUEUE
        );

        $supplierRepository = new SupplierRepository();
        $dataCache          = new DataCache();

        // 全量更新精选供应商索引
        $this->fullSupplierBoutique($queue, $dataCache);

        $totalCount = $supplierRepository->getSupplierCountByDate();
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";

        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierRepository->getAllSupplierId($lastSupplierId,
                self::MAX_COUNT_PER_QUERY);

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values   = implode(self::DELIMITER, $changedSupplierIds);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
                && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $i) {
                echo '平台供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新精选供应商索引
     * @param int $incrementTime
     * @throws \yii\base\InvalidConfigException
     */
    private function incrementSupplierBoutique($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $supplierBoutiqueRepository = new SupplierBoutiqueRepository();
        $totalCount = $supplierBoutiqueRepository->getSupplierCountByTimeSlot($startDate, $endDate);
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";

        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierBoutiqueRepository->getChangedSupplierInfoByTimeSlot($startDate, $endDate,
                $lastSupplierId, self::MAX_COUNT_PER_QUERY);

            if (empty($changedSupplierIds)) {
                continue;
            }

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values   = implode(self::DELIMITER, $changedSupplierIds);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $i) {
                echo '平台精选供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新供应商索引
     * @param int $incrementTime
     * @throws \yii\base\InvalidConfigException
     */
    public function increment($incrementTime = 2)
    {
        /** @var iQueue $queue */
        $queue = \yii::$app->iQueue->createClient(
            QueueConstants::COMPANY_SUPPLIER_INDEX_QUEUE
        );

        $supplierRepository = new SupplierRepository();
        $dataCache          = new DataCache();

        // 返回上一次查询时间
        list($startDate, $endDate) = $dataCache->getLastSearchTime(self::LATEST_SUPPLIER_QUERY_TIME_KEY, $incrementTime);

        // 增量更新精选供应商索引
        $this->incrementSupplierBoutique($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【产品】修改时间
        $this->incrementSupplierProduct($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【案例】修改时间
        $this->incrementSupplierProject($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【品牌词】修改时间
        $this->incrementSupplierBrand($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【联系人表】的修改时间
        $this->incrementSupplierContact($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据用户【认证表】的修改时间uuc_user_identity
        $this->incrementUserIdentity($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【服务分类】的修改时间
        $this->incrementSupplierCategory($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商——根据供应商【报名招募】的报名时间
        $this->incrementSupplierRecruit($queue, $dataCache, $startDate, $endDate);
        /**
         * 20230127 新增
         */
        // 增量更新平台供应商--根据【战略案例（战采数量）】的数量
        $this->incrementSupplierStrategicCase($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商--根据【是否精选供应商】
        $this->incrementSelectionSupplier($queue, $dataCache, $startDate, $endDate);
        // 增量更新平台供应商--根据【营业额】的数量
        $this->incrementSelectionSupplierRevenue($queue, $dataCache, $startDate, $endDate);

        // 增量更新平台供应商--根据【峰会数据】的数量,只需要全量更新一次
        // $this->incrementFhSupplier($queue, $dataCache, $startDate, $endDate);

        $totalCount = $supplierRepository->getSupplierCountByTimeSlot($startDate, $endDate);
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";

        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierRepository->getChangedSupplierInfoByTimeSlot($startDate, $endDate,
                $lastSupplierId, self::MAX_COUNT_PER_QUERY);

            if (empty($changedSupplierIds)) {
                continue;
            }

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values   = implode(self::DELIMITER, $changedSupplierIds);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
                && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $i) {
                echo '平台供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新平台供应商——根据供应商产品修改时间
     * @param string $startDate 开始时间
     * @param string $endDate
     * */
    public function incrementSupplierProduct($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $productRepository = new ProductRepository();
        $supplierIdList = $productRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【产品】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新平台供应商——根据供应商案例修改时间
     * @param string $startDate 开始时间
     * @param string $endDate
     * */
    public function incrementSupplierProject($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $supplierProjectRepository = new SupplierProjectRepository();
        $supplierIdList = $supplierProjectRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【案例】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新平台供应商——根据供应商品牌词修改时间
     * @param string $startDate 开始时间
     * @param string $endDate
     * */
    public function incrementSupplierBrand($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $supplierBrandRepository = new SupplierBrandRepository();
        $supplierIdList = $supplierBrandRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【品牌词】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新平台供应商——根据供应商联系人表的修改时间
     * @param string $startDate 开始时间
     * @param string $endDate
     * */
    public function incrementSupplierContact($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $supplierContactRepository = new SupplierContactRepository();
        $supplierIdList = $supplierContactRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【联系人】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新平台供应商——根据供应商【服务分类】的修改时间
     * @param string $startDate 开始时间
     * @param string $endDate
     * */
    public function incrementSupplierCategory($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $supplierCategoryRepository = new SupplierCategoryRepository();
        $supplierIdList = $supplierCategoryRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【服务分类】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    public function incrementSupplierRecruit($queue, DataCache $dataCache, string $startDate, string $endDate)
    {
        $supplierRecruitRepository = new SupplierRecruitRepository();
        $supplierIdList = $supplierRecruitRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $queue->sendMsg($message, function ($msgBody) use($loopCount, $key) {
                echo '平台供应商【报名招募数据】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 重发机制
     */
    public function resend()
    {
        $this->processResend(
            QueueConstants::COMPANY_SUPPLIER_INDEX_QUEUE
            , self::MODULE
            , '平台供应商'
        );
    }

    /**
     * 清除已经消费过的队列数据
     * @throws \yii\base\InvalidConfigException
     */
    public function clearConsumerQueue()
    {
        $this->processclearConsumerQueue(self::MODULE);
    }

    /**
     * 平台供应商数据加载 just for 测试!!!!!!!!!!!
     */
    public function fulltest()
    {
        /** @var iQueue $queue */
        $queue = \yii::$app->iQueue->createClient(
            QueueConstants::COMPANY_SUPPLIER_INDEX_QUEUE
        );

        $supplierRepository = new SupplierRepository();
        $dataCache          = new DataCache();

        $totalCount = $supplierRepository->getSupplierCountByDate();
        echo "\ntotal supplier: {$totalCount}\n";
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";
        $consumerService = new SupplierConsumerService();

        try {
            for ($i = 0; $i < $loopCount; $i++) {
                $waitUpdateSupplierIds = $supplierRepository->getAllSupplierId($lastSupplierId,
                    self::MAX_COUNT_PER_QUERY);
                $lastSupplierId = $waitUpdateSupplierIds[count($waitUpdateSupplierIds) - 1];


                if (!empty($waitUpdateSupplierIds)) {
                    $start = 0;
                    $sizeCount = (self::MAX_COUNT_PER_INDEX / self::MAX_COUNT_PER_QUERY);

                    $needUpdateIndexSuppliers = [];
                    $needDeleteIndexSuppliers = [];
                    while (1) {

                        $start++;

                        $startTime = explode(' ', microtime());

                        $slicceUpdateSupplierIds = array_slice($waitUpdateSupplierIds, ($start - 1) * $sizeCount, $sizeCount);

                        if (!empty($slicceUpdateSupplierIds)) {


                            $supplierBasics = $supplierRepository->getSupplierBasicInfoByIds($slicceUpdateSupplierIds);
                            if (is_array($supplierBasics)) {
                                $basicIds = array_column($supplierBasics, 'supplier_id');
                                // 判断是否有删除的供应商数据
                                foreach ($slicceUpdateSupplierIds as $id) {
                                    if (!in_array($id, $basicIds)) {
                                        array_push($needDeleteIndexSuppliers, $id);
                                    }
                                }
                            }

                            $consumerService->getSupplierDocumentArray($supplierBasics, $needUpdateIndexSuppliers, $needDeleteIndexSuppliers);

                            $endTime = explode(' ', microtime());
                            $thisTime = $endTime[0] + $endTime[1] - ($startTime[0] + $startTime[1]);
                            echo "平台供应商-增量索引更新：每" . $sizeCount . "条记录执行数据拼接耗时：" . round($thisTime, 3) . " 秒" . PHP_EOL;

                            echo "\nrefresh index...\n";
                            //刷新搜索引擎
                            $consumerService->updateIndex($needUpdateIndexSuppliers, null);
                            $consumerService->deleteIndex($needDeleteIndexSuppliers, null);

                            $needUpdateIndexSuppliers = [];
                            $needDeleteIndexSuppliers = [];
                        }else{
                            break;
                        }
                    }
                }
            }
            echo "\n full test done...";
        }catch (\Throwable $e){
            echo "\nerror: " . $e->getMessage();
        }
    }


    /**
     * 精选供应商库内战略案例
     * @param $queue
     * @param DataCache $dataCache
     * @param string $startDate
     * @param string $endDate
     * @return void
     */
    private function incrementSupplierStrategicCase($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $supplierRecruitRepository = new SelectionSupplierProjectRepository();
        $supplierIdList = $supplierRecruitRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $this->sendQueueMsg($queue,  $dataCache, $supplierIdList, '战略案例');
    }


    /**
     * 精选供应商
     * @param $queue
     * @param DataCache $dataCache
     * @param string $startDate
     * @param string $endDate
     * @return void
     */
    private function incrementSelectionSupplier($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $supplierRecruitRepository = new SelectionSupplierRepository();
        $supplierIdList = $supplierRecruitRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $this->sendQueueMsg($queue,  $dataCache, $supplierIdList, '精选供应商');
    }


    /**
     * 精选-营业额
     * @param $queue
     * @param DataCache $dataCache
     * @param string $startDate
     * @param string $endDate
     * @return void
     */
    private function incrementSelectionSupplierRevenue($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $supplierRecruitRepository = new SelectionSupplierRevenueRepository();
        $supplierIdList = $supplierRecruitRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $this->sendQueueMsg($queue,  $dataCache, $supplierIdList, '营业额');
    }

    /**
     * 峰会数据
     * @param $queue
     * @param DataCache $dataCache
     * @param string $startDate
     * @param string $endDate
     * @return void
     */
    private function incrementFhSupplier($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $supplierRecruitRepository = new FhSupplierMulitDetailRepository();
        $supplierIdList = $supplierRecruitRepository->getSupplierIdsByTimeSlot($startDate, $endDate);
        $this->sendQueueMsg($queue,  $dataCache, $supplierIdList, '峰会数据');
    }


    /**
     * 认证用户
     * @param $queue
     * @param DataCache $dataCache
     * @param $startDate
     * @param $endDate
     * @return void
     * @throws \yii\base\InvalidConfigException
     */
    private function incrementUserIdentity($queue, DataCache $dataCache, $startDate, $endDate)
    {
        $userService = new UserService();
        $companyIdList = $userService->getUserIdentityByTimeSlot($startDate, $endDate);
        $this->sendQueueMsg($queue,  $dataCache, $companyIdList, '用户认证表');
    }

    /**
     * 写入queue
     * @param $queue
     * @param $supplierIdList
     * @param DataCache $dataCache
     * @return void
     */
    private function sendQueueMsg($queue, DataCache $dataCache, $supplierIdList, $desc){
        $loopCount = ceil(count($supplierIdList) / self::MAX_COUNT_PER_QUERY);
        $supplierIdListChunk = array_chunk($supplierIdList, self::MAX_COUNT_PER_QUERY);
        foreach ($supplierIdListChunk as $key => $item){
            $values   = implode(self::DELIMITER, $item);
            $data     = ['source' => 'company', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在,则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
            && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);
            $queue->sendMsg($message, function ($msgBody) use($desc, $loopCount, $key) {
                echo '平台供应商【' . $desc . '】-已插入队列比例: ' . ($key + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }
}
